package com.dd.stack.kafka.producer.config.kafka;

import com.dd.stack.common.util.annotation.Explanation;
import lombok.Builder;
import lombok.Data;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.config.TopicConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.KafkaAdmin;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author liuxianmeng
 * @CreateTime 2025/6/26 18:15
 * @Description Kafka主题初始化配置
 */
@Configuration
public class KafkaTopicInitConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public KafkaAdmin kafkaAdmin() {
        // 构建KafkaAdmin配置
        Map<String, Object> configs = new HashMap<>();
        configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        // 添加安全配置（如果有）
        // configs.put("security.protocol", "SSL");
        // configs.put("ssl.truststore.location", "/path/to/truststore.jks");
        // configs.put("ssl.truststore.password", "password");

        KafkaAdmin admin = new KafkaAdmin(configs);
        admin.setFatalIfBrokerNotAvailable(true);   // 启动时验证Kafka可用性
        // TODO 解决此版本配置问题
        //admin.setModifyTopicConfigs(true);          // 允许更新主题配置
        admin.setAutoCreate(false);                 // 禁用自动创建
        return admin;
    }

    /************************ 初始化主题 START ***********************/

    @Bean
    public NewTopic testTopic() {
        return buildTopic(
                TopicInfo.builder().topicName("test_topic").partitionsNum(2)
                        .replicasNum((short) 3).retentionMs("604800000").build()
        );
    }

    /************************ 初始化主题 END ***********************/

    /**
     * 通用主题构建方法
     * @param topicInfo 主题配置信息（构造）类
     * @return NewTopic
     */
    private NewTopic buildTopic(TopicInfo topicInfo) {
        return TopicBuilder.name(topicInfo.getTopicName())
                .partitions(topicInfo.getPartitionsNum())
                .replicas(topicInfo.getReplicasNum())
                // 消息保留时间 604800000(7Days) 86400000(1Day) 2592000000(30Days)
                .config(TopicConfig.RETENTION_MS_CONFIG, topicInfo.getRetentionMs())
                // 删除策略
                .config(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE)
                .build();
    }

    @Data
    @Builder
    @Explanation("主题配置信息（构造）类")
    private static class TopicInfo {
        @Explanation("主题名")
        private String topicName;
        @Explanation("分区数量")
        private int partitionsNum;
        @Explanation("副本数量")
        private short replicasNum;
        @Explanation("消息保留时间（毫秒）")
        private String retentionMs;
    }
}
